[CELEBORN-2378] Fix CongestionController never resuming due to Netty pool overhead in getTotalPendingBytes() by Deegue · Pull Request #3756 · apache/celeborn

Deegue · 2026-07-10T08:01:35Z

What changes were proposed in this pull request?

Use MemoryManager.instance().getPinnedMemory() for low watermark to exit congestion status when using pooled allocator.

Why are the changes needed?

getMemoryUsage() returns nettyUsedDirectMemory + sortMemory, which includes Netty's pooled allocator overhead.

Change to getPinnedMemory() which returns nettyPinnedDirectMemory + sortMemory instead.

As for unpooled allocator, we keep the same behavior as before.

Does this PR resolve a correctness bug?

Yes, the worker is supposed to resume when pending data drops below the low watermark.

Does this PR introduce any user-facing change?

No API change.
Workers with congestion control and pooled allocator will now correctly resume from congestion state when actual pending data drops below the low watermark.

How was this patch tested?

UTs

Copilot

Pull request overview

Updates worker congestion-control pending-bytes calculation to avoid Netty pooled allocator overhead so the worker can correctly resume once actual pending data drops below the low watermark.

Changes:

Switch CongestionController.getTotalPendingBytes() from MemoryManager.getMemoryUsage() to MemoryManager.getPinnedMemory().

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

  public long getTotalPendingBytes() {
-    return MemoryManager.instance().getMemoryUsage();
+    return MemoryManager.instance().getPinnedMemory();
  }


  public long getTotalPendingBytes() {
-    return MemoryManager.instance().getMemoryUsage();
+    return MemoryManager.instance().getPinnedMemory();
  }


leixm · 2026-07-10T09:42:39Z

It's ok for me, but if we change getMemoryUsage to getPinnedMemory, we should instruct users in migration.md to lower the watermark to avoid off-heap OOM (Out of Memory) errors. Also, we should address the issue of calling getPinnedMemory at 10ms intervals. cc @RexXiong WDYT.

zaynt4606

In non-POOLED mode (UNPOOLED or ADAPTIVE), pooledByteBufAllocators stays empty because both fill sites guard with if (allocatorType == POOLED).
So getNettyPinnedDirectMemory() sums an empty list and returns 0, making getTotalPendingBytes() underestimate pending data.

Suggested fix:
fall back to PlatformDependent.usedDirectMemory() when no pooled allocators exist, so non-POOLED configs retain the old behavior.

Deegue · 2026-07-14T09:20:31Z

Thanks @zaynt4606 , added a fallback when no pooled allocators exist, we fall back to getMemoryUsage().

Deegue · 2026-07-14T09:29:26Z

I also changed the code to keep the enter condition using getTotalPendingBytes() which returns getMemoryUsage() as before. So this fix is only targeting to exit condition with pool allocator.

Gentle ping @leixm @zaynt4606 for further ideas.

Copilot

Pull request overview

Copilot reviewed 1 out of 1 changed files in this pull request and generated 3 comments.

+  public long getActivePendingBytes() {
+    MemoryManager memoryManager = MemoryManager.instance();
+    long pinnedMemory = memoryManager.getPinnedMemory();
+    if (pinnedMemory == 0 && NettyUtils.getAllPooledByteBufAllocators().isEmpty()) {
+      return memoryManager.getMemoryUsage();
+    }
+    return pinnedMemory;
+  }



  protected void checkCongestion() {
    try {
      long pendingConsume = getTotalPendingBytes();


+            logger.info(
+                "Pending consume and produce speed is lower than low watermark, exit congestion control");
+          }


… getTotalPendingBytes()

Deegue · 2026-07-14T14:02:03Z

Offline discussed with @leixm , added cache and update interval for getPinnedMemory to avoid rapid call during congestion status.

SteNicholas · 2026-07-15T02:40:57Z

@Deegue, please discuss this in comments, instead of offline discussion. Otherwise, reviewers and others could not know the input of this change. BTW, please update the title and description according to updates.

Deegue · 2026-07-15T03:24:38Z

@Deegue, please discuss this in comments, instead of offline discussion. Otherwise, reviewers and others could not know the input of this change. BTW, please update the title and description according to updates.

Thanks @SteNicholas, PR description updated.

Previously we discussed if it's necessary to add a rate limit and interval to call getPinnedMemory which iterates all pooled Netty allocators. Since it can add noticeable CPU/alloc overhead at this frequency after hitting the high watermark, we decided to add a cache for that.

Copilot

Pull request overview

Copilot reviewed 1 out of 1 changed files in this pull request and generated 3 comments.

Deegue · 2026-07-15T06:26:23Z

    this.workerSource = workerSource;
    this.sampleTimeWindowSeconds = sampleTimeWindowSeconds;
    this.userInactiveTimeMills = conf.workerCongestionControlUserInactiveIntervalMs();
+    this.activePendingBytesCacheIntervalMs = conf.workerPinnedMemoryCheckIntervalMs();


The behavior of trimMemoryUsage() is controlled by current workerCongestionControlCheckIntervalMs (default 10ms) which this PR doesn't change.

Pinned memory check interval is designed to follow CelebornConf.WORKER_PINNED_MEMORY_CHECK_INTERVAL according to comments above.

+            logger.info(
+                "Pending consume and produce speed is lower than low watermark, exit congestion control");
+          }


+      if (overHighWatermark.get()) {
+        long activePendingBytes = getActivePendingBytes();
+        if (activePendingBytes < workerTrafficQuota.diskBufferLowWatermark()
+            && workerProduceSpeed < workerTrafficQuota.workerProduceSpeedLowWatermark()) {
+          if (overHighWatermark.compareAndSet(true, false)) {


github-actions Bot added the module:worker label Jul 10, 2026

SteNicholas requested a review from Copilot July 10, 2026 08:18

Copilot started reviewing on behalf of SteNicholas July 10, 2026 08:29 View session

Copilot AI reviewed Jul 10, 2026

View reviewed changes

leixm requested review from RexXiong and leixm July 10, 2026 09:43

zaynt4606 reviewed Jul 12, 2026

View reviewed changes

Deegue force-pushed the CELEBORN-2378 branch from 714c10b to 8ba5c68 Compare July 14, 2026 09:15

SteNicholas requested a review from Copilot July 14, 2026 09:30

Copilot started reviewing on behalf of SteNicholas July 14, 2026 09:43 View session

Copilot AI reviewed Jul 14, 2026

View reviewed changes

Deegue force-pushed the CELEBORN-2378 branch from 8ba5c68 to 15db245 Compare July 14, 2026 11:57

Fix CongestionController never resuming due to Netty pool overhead in…

c9ef8ad

… getTotalPendingBytes()

Deegue force-pushed the CELEBORN-2378 branch from 15db245 to c9ef8ad Compare July 14, 2026 12:20

SteNicholas requested a review from Copilot July 15, 2026 03:31

Copilot started reviewing on behalf of SteNicholas July 15, 2026 03:31 View session

Copilot AI reviewed Jul 15, 2026

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

[CELEBORN-2378] Fix CongestionController never resuming due to Netty pool overhead in getTotalPendingBytes()#3756

[CELEBORN-2378] Fix CongestionController never resuming due to Netty pool overhead in getTotalPendingBytes()#3756
Deegue wants to merge 1 commit into
apache:mainfrom
Deegue:CELEBORN-2378

Deegue commented Jul 10, 2026 •

edited

Loading

Uh oh!

Copilot AI left a comment

Uh oh!

leixm commented Jul 10, 2026

Uh oh!

zaynt4606 left a comment

Uh oh!

Deegue commented Jul 14, 2026

Uh oh!

Deegue commented Jul 14, 2026

Uh oh!

Copilot AI left a comment

Uh oh!

Deegue commented Jul 14, 2026

Uh oh!

SteNicholas commented Jul 15, 2026

Uh oh!

Deegue commented Jul 15, 2026

Uh oh!

Copilot AI left a comment

Uh oh!

Deegue Jul 15, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

Uh oh!

Conversation

Deegue commented Jul 10, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR resolve a correctness bug?

Does this PR introduce any user-facing change?

How was this patch tested?

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

leixm commented Jul 10, 2026

Uh oh!

zaynt4606 left a comment

Choose a reason for hiding this comment

Uh oh!

Deegue commented Jul 14, 2026

Uh oh!

Deegue commented Jul 14, 2026

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

Deegue commented Jul 14, 2026

Uh oh!

SteNicholas commented Jul 15, 2026

Uh oh!

Deegue commented Jul 15, 2026

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

Deegue Jul 15, 2026

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

Deegue commented Jul 10, 2026 •

edited

Loading